package com.cloud.rou.im.router;

import com.cloud.rou.im.cluster.naming.node.ImServerInstance;
import com.cloud.rou.im.cluster.naming.node.InstanceParser;
import com.cloud.rou.im.cluster.zk.ZookeeperClientFactory;
import com.cloud.rou.im.protocol.serialize.FastJsonSerializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;

import java.io.IOException;
import java.util.Optional;

/**
 * <p></p>
 *
 * @author GaoShuangWen
 * @since 2019-11-14 21:19
 */
@Slf4j
public class RouterServer {



    public static void main(String[] args) throws IOException {
        CustomInstancePostProcessorAdapt processor = new CustomInstancePostProcessorAdapt();
        DefaultByteToInstanceParser parser= new DefaultByteToInstanceParser();
      /*  SubInstanceEvent subInstanceEvent = new SubInstanceEvent();
        subInstanceEvent.sub(new InstanceTreeCacheListener());*/
         CuratorFramework client = ZookeeperClientFactory.getInstance().getZkCli();
        try {
            //订阅节点的增加和删除事件
            TreeCache treeCache = new TreeCache(client, "/im/instance");
            TreeCacheListener l = new TreeCacheListener() {
                @Override
                public void childEvent(CuratorFramework client,
                                       TreeCacheEvent event) throws Exception {
                    ChildData data = event.getData();
                    if (data != null) {

                        Optional<ImServerInstance> parse = parser.parse(data);
                        switch (event.getType()) {
                            case NODE_REMOVED:
                               // System.out.println("移除"+data.getPath());
                                processor.instanceRemovedProcessor(new ImServerInstance(getId(data.getPath())));
                                break;
                            case NODE_ADDED:
                                parse.ifPresent(processor::instanceAddedProcessor);
                                break;
                            default:
                                break;
                        }
                    } else {
                        log.info("[TreeCache]节点数据为空, path={}");
                    }
                }
            };
            treeCache.getListenable().addListener(l);
            treeCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }

        System.out.println("路由启动成功！");
        // 阻塞当前线程
        System.in.read();
    }
    public  static class DefaultByteToInstanceParser implements InstanceParser<ChildData> {
        @Override
        public Optional<ImServerInstance> parse(ChildData data) {
            byte[] payload = data.getData();
            if(payload!=null&&payload.length>0){
                ImServerInstance deserialize = FastJsonSerializer.getInstance().deserialize(payload, ImServerInstance.class);
                deserialize.setId(getId(data.getPath()));
                return Optional.of(deserialize);
            }
            return Optional.empty();
        }
    }

    /**
     * 取得IM 节点编号
     *
     * @return 编号
     */
    public static long getId(String pathRegistered) {
        String sid = null;
        if (null == pathRegistered) {
            throw new RuntimeException("节点注册失败");
        }
        int index = pathRegistered.lastIndexOf("/im/instance/id-");
        if (index >= 0) {
            index += "/im/instance/id-".length();
            sid = index <= pathRegistered.length() ?
                    pathRegistered.substring(index) : null;
        }
        if (null == sid) {
            throw new RuntimeException("节点ID生成失败");
        }
        return Long.parseLong(sid);
    }
}
